#include "sys.h"
#include "net.h"
#include "log.h"

#include "crc32.h"


#define KAFKA_CLI_ID "rdKafka"

#define MAX_BUFFER_SIZE (1024 * 4)
#define MAX_MDATA_BORKERS_COUNT 20
#define MAX_MDATA_TOPICS_COUNT 4
#define MAX_MDATA_PATIONS_COUNT 32
#define MAX_COMM_COUNT 32

#define KAFKA_API_PRODUCE_REQ 0
#define KAFKA_API_METADATA_REQ 3

#if 0
#define DEBUG(fmt, args...) log_trace(fmt, ##args)
#else
#define DEBUG(fmt, args...)
#endif

typedef struct borker_info_s {
    int id;
    char host[128];
    int port;
} borker_info_t;

typedef struct {
    short errcode;
    int id;
    int leader;

    int replicas_count;
    int replicas[64];

    int lsr_count;
    int lsr[64];
} partion_info_t;
typedef struct {
    short errcode;
    char topic[32];

    int partion_count;
    partion_info_t partions[MAX_MDATA_PATIONS_COUNT];
} topic_info_t;

typedef struct {

    int borkers_count;
    borker_info_t borkers[MAX_MDATA_BORKERS_COUNT];

    int topics_count;
    topic_info_t topics[MAX_MDATA_TOPICS_COUNT];
} mdata_resp_info_t;



typedef struct kafka_comm_hdr_s {
    int len;
    short api_key;
    short api_version;
    int corr_id;
    short client_id_len;
    char* client_id;
} kafka_comm_hdr_t;

uint64_t htonll(uint64_t val)
{
    uint32_t high, low;
    uint64_t ret = 0;

    low = val & 0xffffffff;
    high = (val >> 32) & 0xffffffff;

    low = htonl(low);
    high = htonl(high);

    // printf("htonll low:%x %d high:%x\n", low, low, high);
    ret = low;
    ret = ret << 32;
    ret |= (high);
    // printf("htonll val:%llx ret:%llx, sizeof(ret):%d\n", val, ret, sizeof(ret));
    return ret;
}

static int read_int32(char* ptr, int* value)
{
    *value = ntohl(*((int*)ptr));
    return sizeof(int);
}
static int read_int16(char* ptr, short* value)
{
    *value = ntohs(*((short*)ptr));
    return sizeof(short);
}
static int read_str(char* ptr, char* buf, int len)
{
    strncpy(buf, ptr, len);
    return len;
}

static int write_int32(char* ptr, int value)
{
    *((int*)ptr) = htonl(value);
    return sizeof(int);
}
static int write_int16(char* ptr, short value)
{
    *((short*)ptr) = htons(value);
    return sizeof(short);
}
static int write_int8(char* ptr, char value)
{
    *ptr = value;
    return sizeof(char);
}
static int write_int64(char* ptr, long long value)
{
    long long val = htonll(value);
    memcpy(ptr, &val, sizeof(val));
    return sizeof(long long);
}
static int write_str(char* ptr, char* buf, int len)
{
    strncpy(ptr, buf, len);
    return len;
}
static void get_addrinfo(char* borkers, struct sockaddr_in* addr)
{
    char *delim = ":", *_borkers = NULL, *saveptr = NULL;
    char* ptr;

    _borkers = strdup(borkers);
    ptr = strtok_r(_borkers, delim, &saveptr);
    addr->sin_addr.s_addr = wd_ipaddr(ptr);

    ptr = strtok_r(NULL, delim, &saveptr);
    addr->sin_port = htons(atoi(ptr));

    free(_borkers);
}
static void set_comhdr_len(char* buf, int len)
{
    *((int*)buf) = htonl(len - sizeof(int)); // sizeof(int) is kcom_hdr->len size.
}
static int set_comhdr(char* buf, short api_key, short api_version, int corr_id)
{
    int len = 0;
    kafka_comm_hdr_t* kcom_hdr;

    kcom_hdr = (kafka_comm_hdr_t*)buf;
    kcom_hdr->len = 0;
    len += sizeof(kcom_hdr->len);

    kcom_hdr->api_key = htons(api_key);
    len += sizeof(kcom_hdr->api_key);

    kcom_hdr->api_version = htons(api_version);
    len += sizeof(kcom_hdr->api_version);

    kcom_hdr->corr_id = htonl(corr_id);
    len += sizeof(kcom_hdr->corr_id);

    kcom_hdr->client_id_len = htons(strlen(KAFKA_CLI_ID));
    len += sizeof(kcom_hdr->client_id_len);

    kcom_hdr->client_id = (char*)kcom_hdr + len;
    strcpy(kcom_hdr->client_id, KAFKA_CLI_ID);
    len += strlen(KAFKA_CLI_ID);

    return len;
}
static int _deal_partions(char* buf, topic_info_t* topic)
{
    char *ptr = NULL, *sptr = NULL;
    int i, idx, j, replicas_idx, n, lsr_idx;

    sptr = ptr = buf;
    ptr += read_int32(ptr, &topic->partion_count);
    DEBUG("topic->partion_count:%d", topic->partion_count);
    for (i = 0, idx = 0; i < topic->partion_count; i++, idx++) {
        if (idx > MAX_MDATA_PATIONS_COUNT)
            idx = 0;

        ptr += read_int16(ptr, &topic->partions[idx].errcode);
        DEBUG("topic->partions[%d].errcode:%d", idx, topic->partions[idx].errcode);
        ptr += read_int32(ptr, &topic->partions[idx].id);
        DEBUG("topic->partions[%d].id:%d", idx, topic->partions[idx].id);
        ptr += read_int32(ptr, &topic->partions[idx].leader);
        DEBUG("topic->partions[%d].leader:%d", idx, topic->partions[idx].leader);

        ptr += read_int32(ptr, &topic->partions[idx].replicas_count);
        DEBUG("topic->partions[%d].replicas_count:%d", idx, topic->partions[idx].replicas_count);
        for (j = 0, replicas_idx = 0; j < topic->partions[idx].replicas_count; j++, replicas_idx++) {
            if (j > MAX_COMM_COUNT)
                replicas_idx = 0;
            ptr += read_int32(ptr, &topic->partions[idx].replicas[replicas_idx]);
            DEBUG("topic->partions[%d].replicas[%d]:%d", idx, replicas_idx, topic->partions[idx].replicas[replicas_idx]);
        }
        ptr += read_int32(ptr, &topic->partions[idx].lsr_count);
        DEBUG("topic->partions[%d].lsr_count:%d", idx, topic->partions[idx].lsr_count);
        for (n = 0, lsr_idx = 0; n < topic->partions[idx].replicas_count; n++, lsr_idx++) {
            if (n > MAX_COMM_COUNT)
                lsr_idx = 0;
            ptr += read_int32(ptr, &topic->partions[idx].lsr[lsr_idx]);
            DEBUG("topic->partions[%d].lsr[%d]:%d", idx, lsr_idx, topic->partions[idx].lsr[lsr_idx]);
        }
    }
    return (ptr - sptr);
}

static int _deal_metadata_response(char* buffer, int len, mdata_resp_info_t* mdata)
{
    char *ptr = NULL, *sptr = NULL;
    int i, idx;
    short host_len, topic_len;
    int resp_len, corr_id;
    int tmp_borkers_count = 0;

    sptr = ptr = buffer;
    ptr += read_int32(ptr, &resp_len);
    DEBUG("resp_len:%d", resp_len);
    ptr += read_int32(ptr, &corr_id);
    DEBUG("corr_id:%d", corr_id);

    ptr += read_int32(ptr, &mdata->borkers_count);
    DEBUG("mdata->borkers_count:%d", mdata->borkers_count);

    if(mdata->borkers_count > MAX_MDATA_BORKERS_COUNT){
        tmp_borkers_count = MAX_MDATA_BORKERS_COUNT;
    }else{
        tmp_borkers_count = mdata->borkers_count;
    }
    // for (i = 0, idx = 0; i < mdata->borkers_count; i++, idx++) {
    for (i = 0, idx = 0; i < tmp_borkers_count; i++, idx++) {
        if (idx > MAX_MDATA_BORKERS_COUNT)
            idx = 0;

        ptr += read_int32(ptr, &mdata->borkers[idx].id);
        DEBUG("mdata->borkers[%d].id:%d", idx, mdata->borkers[idx].id);
        ptr += read_int16(ptr, &host_len);
        DEBUG("[%d]host_len:%d", idx, host_len);
        ptr += read_str(ptr, mdata->borkers[idx].host, host_len);
        DEBUG("mdata->borkers[%d].host:%s", idx, mdata->borkers[idx].host);
        ptr += read_int32(ptr, &mdata->borkers[idx].port);
        DEBUG("mdata->borkers[%d].port:%d", idx, mdata->borkers[idx].port);
    }

    ptr += read_int32(ptr, &mdata->topics_count);
    DEBUG("mdata->topics_count:%d", mdata->topics_count);
    for (i = 0, idx = 0; i < mdata->topics_count; i++, idx++) {
        if (idx > MAX_MDATA_TOPICS_COUNT)
            idx = 0;

        ptr += read_int16(ptr, &mdata->topics[idx].errcode);
        DEBUG("mdata->topics[%d].errcode:%d", idx, mdata->topics[idx].errcode);
        ptr += read_int16(ptr, &topic_len);
        DEBUG("[%d]topic_len:%d", idx, topic_len);
        ptr += read_str(ptr, mdata->topics[idx].topic, topic_len);
        DEBUG("mdata->topics[%d].topic:%s", idx, mdata->topics[idx].topic);

        ptr += _deal_partions(ptr, &mdata->topics[idx]);
    }
    return (ptr - sptr);
}
static int send_metadata(int fd, char* topic, mdata_resp_info_t* mdata)
{
    int len = 0, ret;
    char *buffer, *ptr = NULL;

    buffer = calloc(1, MAX_BUFFER_SIZE);

    ptr = buffer;
    len = set_comhdr(ptr, KAFKA_API_METADATA_REQ, 0, 1);
    ptr += len;

    //set topic
    ret = write_int32(ptr, 1);
    ptr += ret;
    len += ret;
    ret = write_int16(ptr, strlen(topic));
    ptr += ret;
    len += ret;
    ret = write_str(ptr, topic, strlen(topic));
    len += ret;

    //total len
    set_comhdr_len(buffer, len);

    ret = write(fd, buffer, len);
    if (ret < 0) {
        printf("%s:%d write failed\n", __FUNCTION__, __LINE__);
        ret = -1;
        goto out;
    }

    bzero(buffer, MAX_BUFFER_SIZE);
    ret = read(fd, buffer, MAX_BUFFER_SIZE);
    if (ret < 0) {
        printf("%s:%d read failed\n", __FUNCTION__, __LINE__);
        ret = -1;
        goto out;
    }
    DEBUG("response len:%d\n", ret);

    _deal_metadata_response(buffer, ret, mdata);
    ret = 1;
out:
    free(buffer);
    return ret;
}
static topic_info_t* find_topic_info(mdata_resp_info_t* mdata, char* topic)
{
    int i = 0;

    for (i = 0; i < mdata->topics_count; i++) {
        if (strcmp(topic, mdata->topics[i].topic) == 0) {
            return &mdata->topics[i];
        }
    }
    return NULL;
}
static void balance_svr(mdata_resp_info_t* mdata, char* topic, borker_info_t** borker, partion_info_t** partion)
{
    int i;
    int borker_id, partion_idx;
    topic_info_t* topic_info;

    topic_info = find_topic_info(mdata, topic);
    partion_idx = time(0) % topic_info->partion_count;
    *partion = &topic_info->partions[partion_idx];
    
    borker_id = topic_info->partions[partion_idx].leader;
    for(i = 0; i < mdata->borkers_count; i++){
        if(borker_id == mdata->borkers[i].id){
            *borker = &mdata->borkers[i];
            break;            
        }
    }
}
static int mk_msgset(char* buf, char* msg, int msglen)
{
    int len = 0, ret = 0, msg_size = 0;
    uint32_t crc =0;
    char* ptr, *ptr_crc;

    ptr = buf;

    crc = my_crc32_init();

    //offset
    #if 0
    ret = write_int64(ptr, 0);
    #else
    ret = sizeof(uint64_t);
    #endif

    len += ret;
    ptr += ret;
    //message size
    msg_size = sizeof(int) * 3 + sizeof(char)*2 + msglen;
    ret = write_int32(ptr, msg_size);
    len += ret;
    ptr += ret;

    //crc
    ret = write_int32(ptr, 0);
    // DEBUG("crc:%x", crc);
    ptr_crc = ptr;
    len += ret;
    ptr += ret;

    //magic byte
    ret = write_int8(ptr, 0);
    crc = my_crc32_update(crc, ptr, sizeof(char));
    // DEBUG("crc:%x", crc);
    len += ret;
    ptr += ret;
    //attr 
    ret = write_int8(ptr, 0);
    crc = my_crc32_update(crc, ptr, sizeof(char));
    // DEBUG("crc:%x", crc);
    len += ret;
    ptr += ret;

    //key
    ret = write_int32(ptr, -1);
    crc = my_crc32_update(crc, ptr, sizeof(int));
    // DEBUG("crc:%x", crc);
    len += ret;
    ptr += ret;

    //msg 
    ret = write_int32(ptr, msglen);
    crc = my_crc32_update(crc, ptr, sizeof(int));
    // DEBUG("crc:%x", crc);
    len += ret;
    ptr += ret;
    ret = write_str(ptr, msg, msglen);
    crc = my_crc32_update(crc, ptr, msglen);
    // DEBUG("crc:%x", crc);
    len += ret;
    ptr += ret;

    crc = my_crc32_finalize(crc);
    // DEBUG("crc:%x", crc);
    write_int32(ptr_crc, crc);
    return len;
}
static int mk_produce_msg(char* buf,int partion_id, char* topic, int topic_len, char* msg, int msglen)
{
    int len = 0, ret = 0;
    char *ptr, *ptr_msgset_size;

    ptr = buf;
    //set require ack
    ret = write_int16(ptr, 1); // 0:no response, 1:wait server write to log, 2:wait server commit
    len += ret;
    ptr += ret;

    //timeout
    ret = write_int32(ptr, 1000); // ms
    len += ret;
    ptr += ret;

    {
        //topic name
        ret = write_int32(ptr, 1); // topic arr count
        len += ret;
        ptr += ret;
        ret = write_int16(ptr, topic_len); //topic len
        len += ret;
        ptr += ret;
        ret = write_str(ptr, topic, topic_len);
        len += ret;
        ptr += ret;

        {
            //partion
            ret = write_int32(ptr, 1); // partion arr count
            len += ret;
            ptr += ret;
            ret = write_int32(ptr, partion_id);
            len += ret;
            ptr += ret;

            //messageset size
            ptr_msgset_size = ptr;
            ret = write_int32(ptr, 0);
            len += ret;
            ptr += ret;

            //messageset
            ret = mk_msgset(ptr, msg, msglen);
            len += ret;

            //reset messageset size
            write_int32(ptr_msgset_size, ret);
        }
    }
    return len;
}
int deal_produce_response(char *buffer, int buf_len){
    char *ptr;
    short errcode, topic_len;
    int resp_len, count, topics_count,  partions_count, partion, i, j,n;
    char topic[64] = {0};

    ptr = buffer;
    ptr += read_int32(ptr, &resp_len);
    DEBUG("resp_len:%d", resp_len);

    ptr += read_int32(ptr, &count);
    DEBUG("count:%d", count);
    for(i = 0; i < count; i++){
        ptr += read_int32(ptr, &topics_count);
        DEBUG("topics_count:%d", topics_count);
        for(j = 0; j < topics_count; j++){
            ptr += read_int16(ptr, &topic_len);
            DEBUG("topic_len:%d", topic_len);

            bzero(topic, sizeof(topic));
            ptr += read_str(ptr, topic, topic_len);
            DEBUG("topic:%s", topic);

            ptr += read_int32(ptr, &partions_count);
            DEBUG("partions_count:%d", partions_count);
            for(n = 0; n < partions_count; n++){
                ptr += read_int32(ptr, &partion);
                DEBUG("partion:%d", partion);
                ptr += read_int16(ptr, &errcode);
                DEBUG("errcode:%d", errcode);
            }
        }
    }
    DEBUG("errcode:%d", errcode);
    return errcode;
}
int send_produce(borker_info_t* borker, partion_info_t* partion, char* topic, int topic_len, char* msg, int msglen)
{
    int fd = 0;
    int len = 0, ret = 0, errcode = 0;
    unsigned short port;
    char *buffer, *ptr;
    struct sockaddr_in addr;

    log_trace("msglen:%d", msglen);
    
    buffer = calloc(1, msglen + 512);
    if (!buffer) {
        DEBUG("calloc buffer failed.");
        ret = -1;
        goto out;
    }
    ptr = buffer;

    fd = socket(AF_INET, SOCK_STREAM, 0);
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = wd_ipaddr(borker->host);
    port = borker->port;
    addr.sin_port = htons(port);

    if (0 > connect(fd, (struct sockaddr*)&addr, sizeof(addr))) {
        DEBUG("connect error.");
        ret = -1;
        goto out;
    }

    len = set_comhdr(ptr, KAFKA_API_PRODUCE_REQ, 1, 1);
    ptr += len;

    len += mk_produce_msg(ptr, partion->id, topic, topic_len, msg, msglen);

    set_comhdr_len(buffer, len);

    if( 0 > write(fd, buffer, len)){
        DEBUG("write error.");
        ret = -1;
        goto out;
    }

    ret = read(fd, buffer, MAX_BUFFER_SIZE);
    if(ret < 0){
        DEBUG("read error.");
        ret = -1;
        goto out;
    }
    DEBUG("read len:%d", ret);

    if(deal_produce_response(buffer, ret) > 0){
        ret = -1;
        goto out;
    }

out:
    if(fd > 0){
        close(fd);
    }
    if(buffer)
        free(buffer);
    return ret;
}
int simpleKafka_send(char *borker, char *topic, char *msg, int msg_len)
{
    int fd, ret, i;
    struct sockaddr_in addr;
    mdata_resp_info_t mdata_resp = { 0 };
    borker_info_t* sub_borker;
    partion_info_t* partion;

    addr.sin_family = AF_INET;
    get_addrinfo(borker, &addr);

    if(addr.sin_addr.s_addr == 0){
        return -1;
    }

    fd = socket(AF_INET, SOCK_STREAM, 0);

    ret = connect(fd, (struct sockaddr*)&addr, sizeof(addr));
    if (ret < 0) {
        log_trace("simple kafka connect error.");
        log_syserr("connect error.");
        close(fd);
        return -1;
    }

    ret = send_metadata(fd, topic, &mdata_resp);
    if(ret < 0){
        close(fd);
        return -1;
    }
    close(fd);

#if 1
    balance_svr(&mdata_resp, topic, &sub_borker, &partion);
    DEBUG("host:%s:%d, partion id:%d", sub_borker->host, (sub_borker->port), partion->id);
#else
    borker_info_t tmp_borker = {
        .id = 0,
        .host = "101.207.125.93",
        .port = 9092,
    };
    sub_borker = &tmp_borker;
    partion_info_t tmp_partion = {
        .id = 0,
    };
    partion = &tmp_partion;
#endif
    ret = send_produce(sub_borker, partion, topic, strlen(topic), msg, msg_len);

    return ret;
}

int skafak_send(char *borkers, char *topic, char *msg, int msg_len){
    char *delim = ",", *ptr = NULL, *saveptr=NULL;
    char *tmp_borkers, *tmp_ptr;
    int ret = 0;

    log_trace("borkers:%s", borkers);
    tmp_ptr = tmp_borkers = strdup(borkers);

    while((ptr = strtok_r(tmp_ptr, delim, &saveptr)) != NULL){
        ret = simpleKafka_send(ptr, topic, msg, msg_len);
        log_trace("ret:%d, kafka host:%s",ret, ptr);
        if(ret > 0)
            goto out;
        tmp_ptr = NULL;
    }
out:
    if(ret > 0){
        log_trace("simple kafak send to %s SUCCESS.", ptr);
    }else{
        log_trace("simple kafak all send failed(%s).", borkers);
    }
    free(tmp_borkers);
    return ret;
}

#if 0
int main(int argc, char **argv){
    char* Borkers = "123.147.221.146:19001";
    char* Topic = "netprobeV2";
    char msg[] = "asdfasdfasdf";
    simpleKafka_send(Borkers, Topic, msg, strlen(msg));
}
#endif